package com.ask.forMe.mq.consumer;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.ask.forMe.model.message.UserBehaviorEvent;
import com.ask.forMe.model.entity.Rating;
import com.ask.forMe.service.RatingService;
import com.ask.forMe.service.RatingUpdateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;

@Service
@Slf4j
public class UserBehaviorConsumer {
    @Autowired
    private RatingService ratingService;
    @Autowired
    private RatingUpdateService ratingUpdateService;


    /**
     * 消费者，监听用户的行为事件
     *
     * @param event
     */
    //@KafkaListener(topics = "user-behaviors", groupId = "my-group")
    public void consume(UserBehaviorEvent event) {
        log.info("Received behavior event: userId={}, itemId={}, type={}",
                event.getUserId(), event.getItemId(), event.getType());

        Rating rating = ratingService.getByUserItemId(event.getUserId(), event.getItemId());

        boolean exist = true;
        if (rating == null) {
            rating = new Rating(event.getUserId(), event.getItemId(), 0.0);
            exist = false;
        }


        // 应用时间衰减
        ratingUpdateService.applyTimeDecay(rating);

        // 处理购买行为
        switch (event.getType()) {
            case PURCHASE:
                ratingUpdateService.updateForPurchase(rating);
                break;
            case FAVORITE:
                ratingUpdateService.updateForFavorite(rating);
                break;
        }
        // 保存跟新后的评分
        if (exist)
            ratingService.updateByUserItemId(rating.getUserId(), rating.getItemId(), rating.getRating());
        else
            ratingService.save(rating);
    }
}
